package com.eternal.base.data.ble;

import com.clj.fastble.data.BleDevice;
import com.eternal.base.utils.ByteUtils;
import com.eternal.framework.utils.KLog;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;

public class ObservableMessage implements Message, ObservableOnSubscribe<byte[]> {
    private ObservableEmitter<byte[]> emitter;
    private Observable<byte[]> observable;
    private byte[] packet;

    ObservableMessage(byte[] packet) {
        this.observable = Observable.create(this);
        this.packet = packet;
    }

    public void onNext(byte[] data) {
        if (!isDisposed()) {
            emitter.onNext(data);
        }
    }

    public void onErr(Throwable throwable) {
        if (!isDisposed()) {
            emitter.onError(throwable);
        }
    }

    public void onComplete() {
        if (!isDisposed()) {
            emitter.onComplete();
        }
    }

    public boolean isDisposed() {
        return emitter == null || emitter.isDisposed();
    }

    @Override
    public byte[] getPacket() {
        return packet;
    }

    @Override
    public Observable<byte[]> getSingle() {
        return observable;
    }

    @Override
    public void subscribe(ObservableEmitter<byte[]> emitter) {
        this.emitter=emitter;
    }

    public Observable<byte[]> reset(){
        observable=Observable.create(this);
        return observable;
    }
}
